package Sun

import java.text.SimpleDateFormat
import java.util.Date
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.mllib.linalg.{DenseVector, Vectors, Vector}
import org.apache.spark.mllib.regression.LabeledPoint

import scala.collection.mutable.ArrayBuffer
import scala.io.Source

/**
 * Created by SHANGMAI on 2016/6/10.
 */
object BasicFunction  extends Serializable{

  //判断模式，单机模型则算法不读redis
  var mode:String = "Redis"

  lazy val  channelSize = {
    if (mode != "StandAlone") {
      val jedis = RedisClient.pool.getResource
      jedis.select(dbIndex)

      //分成不同情况的统计量
      var result = jedis.get("channelcount").toInt
      jedis.close()
      result

    }

    else
    //暂定计算方式
      2010
  }


  //feature中 表示了一个用户的信息  这里主要是表示用户ID
  val info_length = 1


  //feature中表示了时间无关的
  lazy val featureC_length =  {
    if (mode != "StandAlone") {
      val jedis = RedisClient.pool.getResource
      jedis.select(dbIndex)

      //分成不同情况的统计量
      var result = jedis.get("adtypecount").toInt + jedis.get("adwebcount").toInt + channelSize
      jedis.close()
      result
    }

    else
      //暂定计算方式
      channelSize + featureE_length - 3
  }

  //E feature不包括时间和渠道
   lazy val featureE_length = {

    if (mode != "StandAlone") { //通过web和type来进行赋值
    val jedis = RedisClient.pool.getResource
    jedis.select(dbIndex)
    var result = jedis.get("adtypecount").toInt + jedis.get("adwebcount").toInt

    //需要机上三个action的feature
    result = result + 3
    jedis.close()
    result
  }
    else
      78
  }

  //时间差的开始时间计算函数，可能为NOW 可能为其他值
   var startTime:Long = 0


  val featureD_length = 0

  //redis的数据库
  val dbIndex = 1
  
  //weight中的长度

  //规范量纲，从按天计算
  val timedivision = 60*60*1000*24


  //给出几个存储位置，便于中间结果的存放
  //给出个基础文件夹  剩下的都在这上进行修改
  var basicpath:String = "/bigdata/MutiTouchTest"

  //给出原始的的文件地址，adminor的文件名
  lazy val log_path = basicpath + "/public.adm.log.3.0"

  //转换地址的文件名
  lazy val conversion_path = basicpath + "/conversion"

  //渠道的名称
  lazy val appendix_path = basicpath + "/appendix.txt"



  lazy val result_compete_path = basicpath + "/result_all"




  //正例的位置
  lazy val sample_positive_save_path =  basicpath + "/positiveSample"

  //负例的位置
  lazy val sample_negative_save_path = basicpath + "/negativeSample"

  lazy val featureC_path = basicpath + "/featureCresult"
  lazy val channelList_path = basicpath + "/channelList"

  //定义一个存储模型权重的位置
  lazy val modelweight_path = basicpath + "/weight"


  //给出一个读写文件的路径的转换  基本上以spark的存储文件位置为准

  def basicloadpath(loadpath:String): String =
  {
      val result  = loadpath +  "/part-*"
      result
  }

  //只在单机模式的时候 将数据读入到内存
  lazy val channelDic: Map[Long, Int]= {
    if(mode == "StandAlone") {
      var i = 1
      var map = Map[Long, Int]()
      for (line <- Source.fromFile(channelList_path + "/part-00000").getLines){
        map +=  (line.toLong -> i)
        i = i + 1
      }
      map
    }
    else
      null
  }

  //给出返回不同渠道的位置
  def  indexChannel(channel:String) : Int = {

    var result = 0
    var value:Int = 0
    if(mode == "Redis") {
      val jedis = RedisClient.pool.getResource
      jedis.select(dbIndex)
      value = jedis.get("adchannel" + channel).toInt
      jedis.close()
    }

    //如果是单机版，则读取文件
    else
    {
        if (channelDic.contains(channel.toLong))
          value = channelDic(channel.toLong)
    }
    //判断有无匹配
    if (value != 0 && value != null ) {
      //给出相对位置，这里的位置是代表channel的featureE和featureD的位置，一般来说E的位置在D之前  所以这么标记 即该点返回的是第一个E feature的位置
      result =  info_length + featureC_length + (value - 1)* (featureE_length + featureD_length)
    }
    return  result
  }

  //实现一个简单的向量乘法
  def multiplication(vector1: Vector, vector2: Vector): Double = {
    //当数目不一致的的时候报错
    assert(vector1.size == vector2.size)

    var result: Double = 0.0
    for (i <- 0 to vector1.size - 1) {
      result = result + vector1.apply(i) * vector2.apply(i)

    }

    return result
  }

  //做一个向量的乘法
  def multipVector(x:Vector,multinum:Double): Vector =
  {
    var result = ArrayBuffer[Double]()
    for (i <- 0 to x.size - 1)
      result +=  x.apply(i) * multinum

    return  Vectors.dense(result.toArray)

  }

  def alphaFuntion(x: Vector, w: Vector): Double = {

    math.exp(multiplication(w, x))
  }

  //小labda
  def lambdaFuntion1(x: Vector, w: Vector,timediff:Double): Double = {

    alphaFuntion(x,w) * math.exp ( -1 * alphaFuntion(x,w) * timediff)

  }

  //大labda
  def lambdaFuntion2(x: Vector, w: Vector,timediff:Double): Double = {

    1 - math.exp ( -1 * alphaFuntion(w,x) * timediff)

  }


  //进行导数的操作求C的导数
  def derivativeC(x:Vector,w:Vector): Vector =
  {
    val featureC = getStaticFeatureC(x)
    val weightC = getStaticWeightC(w)
    var derivative =  -1 / (1 + logisticFuntion(x,w)) / (1 + logisticFuntion(x,w)) * math.exp(-1 * multiplication(featureC,weightC)) * -1

    return  multipVector(featureC,derivative)
  }

  //基本函数 对于阿尔法求导函数的整理
  def derivativealpha(x:Vector,w:Vector): Vector =
  {
    val derivative = math.exp(multiplication(x,w))
    multipVector(x,derivative)
  }


  //给出计算小labda函数求导
  def derivativelambda1(featureD: Vector, weightD: Vector,timediff:Double): Vector =
  {

    val result  = (1 - lambdaFuntion2(featureD,weightD,timediff)) *  (1 - alphaFuntion(featureD,weightD) * timediff)

    multipVector(derivativealpha(featureD,weightD),result)

  }
  //给出计算大labda函数求导
  def derivativelambda2(featureD: Vector, weightD: Vector,timediff:Double): Vector ={

    var result  = (1 - lambdaFuntion2(featureD,weightD,timediff)) * timediff
    multipVector(derivativealpha(featureD,weightD),result)

  }

  //说明了从复杂的feature和weight中抽取计算的信息
  //该函数是从那个K-V结构中抽取出E的feature 详见S H函数
  def getActionfeatureE(feature: Vector):Vector ={

    var resultBuffer = ArrayBuffer[Double]()

    //E的变量中去掉了时间和渠道
    for (i <- 0 to BasicFunction.featureE_length - 1 ) {
      resultBuffer += feature.apply(i)
    }
    Vectors.dense(resultBuffer.toArray)
  }

  def getActionWeightE(weight: Vector,channel:Double):Vector ={

    var resultBuffer = ArrayBuffer[Double]()
    //获得channel的点的位置 //给出E的起始位置 开始是渠道 时间
    val indexE = indexChannel(channel.toLong.toString)



    //D属性一般排在E后面，
    for (i <- indexE  to indexE + featureE_length - 1) {
      resultBuffer += weight.apply(i)
    }

    Vectors.dense(resultBuffer.toArray)

  }

  def getStaticFeatureC(feature: Vector):Vector = {

    var result = ArrayBuffer[Double]()

    if (feature.size < 500){
      var aad = 423
      aad = aad + 1
    }

    for(i <- info_length  to info_length  + featureC_length - 1)
    {
      result +=  feature.apply(i)
    }
    return Vectors.dense(result.toArray)
  }

  def getStaticWeightC(weight: Vector):Vector ={

    var resultBuffer = ArrayBuffer[Double]()

    //D的变量中从E变量之后开始计算,计算向量
    for (i <- 0 to BasicFunction.featureC_length - 1) {
      resultBuffer += weight.apply(i)
    }

    Vectors.dense(resultBuffer.toArray)

  }

  def getTimeFeatureD(feature: Vector):Vector ={

    var resultBuffer = ArrayBuffer[Double]()

    //D的变量中从E变量之后开始计算,计算向量
    for (i <- BasicFunction.featureE_length  to BasicFunction.featureE_length + BasicFunction.featureD_length - 1) {
      resultBuffer += feature.apply(i)
    }

    Vectors.dense(resultBuffer.toArray)
  }

  def getTimeWeightD(weight: Vector,channel:Double):Vector = {

    var resultBuffer = ArrayBuffer[Double]()
    //获得channel的点的位置
    val indexchannel = indexChannel(channel.toLong.toString)

    //给出E的起始位置 开始是渠道 时间,D延续E
    val indexD = indexchannel  + featureE_length
    //D属性一般排在E后面，
    for (i <- indexD  to indexD + featureD_length - 1) {
      resultBuffer += weight.apply(i)
    }

    Vectors.dense(resultBuffer.toArray)

  }


  def getTime(x: Vector,index:Int):Long ={
    x.apply(index + 20).toLong
  }

//  def logisticFuntion(x: Vector, w: Vector): Double = {
//    1 / (1 + math.exp(multiplication(getStaticFeatureC(x), getStaticWeightC(x))))
//  }

  //给出一个返回C概率的向量
  def logisticFuntion(x:Vector,w:Vector): Double =
  {

    1 / (1 + math.exp(-1 * multiplication(getStaticFeatureC(x),getStaticWeightC(w))) )
  }

  //定义文中的风险函数hazard
  //为聚合好的数据 key(channel,time)  value:feature
  //该数据是经过聚合的一次数据
  def hazardRateFunction(data:Map[(Double,Double),Vector],w:Vector):Double = {
    var result = 1.0

    data.keys.foreach{key =>
      //判断channel向量
        {
          //获取值向量,获取所有的feature
          val x = data(key)
          //获得渠道
          val channel = key._1
          //分开来应对。设置时间
          val timediff = (startTime -   key._2.toDouble) / timedivision
          if(timediff > 0) {
            val alpharesult = alphaFuntion(getActionfeatureE(x), getActionWeightE(w, channel))
            val lambdaresult = lambdaFuntion1(getTimeFeatureD(x), getTimeWeightD(w, channel), timediff)
            result = result + alpharesult * lambdaresult
          }
        }
      }

    return  result
  }

  //传入的数据的格式为((channelid,time),vector)
  def survivalFunction(data:Map[(Double,Double),Vector],w:Vector):Double = {
    var result = 0.0
    data.keys.foreach{key =>{
      //判断channel向量
      //获取值向量
      val x = data(key)
      //获得渠道
      val channel = key._1

      val timediff = (startTime -   key._2.toDouble) / timedivision
      if(timediff > 0) {
        val lambda2Result = lambdaFuntion2(getTimeFeatureD(x), getTimeWeightD(w, channel), timediff)
        val alphaResult = alphaFuntion(getActionfeatureE(x), getActionWeightE(w, channel))
        result = result - alphaResult * lambda2Result
      }
    }}

    //防止溢出
  if (result < -530)
      result = -530

  if(result >  -0.00000001 && result <  0.000000001 )
    {
      if (result < 0)
        result = 0.000000001 * -1

      else
        result = 0.000000001
    }

    return  math.exp(result)
  }

  //进行特征的存储
  def saveSampleData(tempRdd : RDD[((String,String,String), Array[Double])],path:String):Unit =  {

    val FileRdd = tempRdd.map(x => {

      var result:String = ""
      result = result + x._1._1 + ","  + x._1._2 +  ","  + x._1._3
      for(i <- 0 to x._2.length - 1){
        result = result + "," + x._2.apply(i)
      }
      result
    })

    FileRdd.saveAsTextFile(path)
  }

}

